package com.atguigu.upp.apps;

import com.atguigu.upp.beans.TagInfo;
import com.atguigu.upp.constants.TypeCodeConstant;
import com.atguigu.upp.mappers.ClickHouseMapper;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.SqlTaskExecuteUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/9/14
 *
 *  计算后的bitmap表一共有3列:
 *
 *          标签名
 *          标签值
 *          人群(bitmap)
 *
 *    统一使用的是Spark的远超提交器把程序提交到yarn。
 *          spark-yarn/bin/spark-submit  只能提交一个SparkApp
 *
 *          一个SparkApp。必须有SparkContext
 */
public class GenerateBitmapApp
{
    public static void main(String[] args) throws IOException {

        Integer taskId = Integer.parseInt(args[0]);
        String do_date = args[1];
        //String do_date = "2020-06-14";

        //不会用到，要借助spark的提交工具，必须有
        SparkSession sparkSession = SqlTaskExecuteUtil.createSparkSession("GenerateBitmapApp");

        //准备查询mysql的相关工具
        SqlSessionFactory sqlSessionFactory = SqlTaskExecuteUtil.createSqlSessionFactory("mysql-config.xml");
        SqlSession sqlSession = sqlSessionFactory.openSession();
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSession);

        //查询今天已经启用计算的标签
        List<TagInfo> tags = mysqlDBService.getTagInfosWithOn();

        //根据Tag中的 tagValueType对标签进行分类
        List<TagInfo> stringTypeTags = new ArrayList<>();
        List<TagInfo> longTypeTags = new ArrayList<>();
        List<TagInfo> dateTypeTags = new ArrayList<>();
        List<TagInfo> decimalTypeTags = new ArrayList<>();

        for (TagInfo tag : tags) {

            switch (tag.getTagValueType()){
                case TypeCodeConstant.TAG_VALUE_TYPE_DATE : dateTypeTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_LONG :  longTypeTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_STRING :  stringTypeTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_DECIMAL :  decimalTypeTags.add(tag); break;
            }

        }

        /*
                根据每个集合中不同类型的标签，生产sql语句，写入到对应的表中

                stringTypeTags： [   tag_person_nature_gender , tag_person_nature_period        ]

                insert into table user_tag_value_string
                select xxxx
         */

        //获取ClickHouseMapper clickHouseMapper
        SqlSessionFactory ckSSF = SqlTaskExecuteUtil.createSqlSessionFactory("clickhouse-config.xml");
        SqlSession ckSS = ckSSF.openSession();
        ClickHouseMapper clickHouseMapper = ckSS.getMapper(ClickHouseMapper.class);

        //执行写入
        insertIntoBitMap(stringTypeTags,do_date,"user_tag_value_string",clickHouseMapper);
        insertIntoBitMap(longTypeTags,do_date,"user_tag_value_long",clickHouseMapper);
        insertIntoBitMap(dateTypeTags,do_date,"user_tag_value_date",clickHouseMapper);
        insertIntoBitMap(decimalTypeTags,do_date,"user_tag_value_decimal",clickHouseMapper);


    }

    public static void insertIntoBitMap(List<TagInfo> tags, String do_date, String targetTable, ClickHouseMapper clickHouseMapper){

        if (tags.size() > 0){
            //生成 tagSql
            String template = "( '%s' , %s  ) ";
            List<String> tagList = tags.stream().map(tag -> String.format(template, tag.getTagCode().toLowerCase(), tag.getTagCode().toLowerCase())).collect(Collectors.toList());
            String tagSql = StringUtils.join(tagList, ',');

            //获取宽表名字
            String upwideprefix = PropertiesUtil.getValue("upwideprefix");
            String tableName = upwideprefix + do_date.replace('-', '_');

            //写入之前做幂等性处理，如果今天分区中已经写入了数据，把今天分区的数据删除
            clickHouseMapper.deleteBitMapByDate(targetTable,do_date);

            //如果有要写入的tag,执行写入
            clickHouseMapper.insertBitmap(targetTable,tagSql,tableName,do_date);

        }
    }
}
